package org.example.frw.mq.rocket;

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.retry.RetryException;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;


@Component
@Slf4j
public class DelayDoubleDelete {
    @Resource
    private CacheDeleteRetry cacheDeleteRetry;

    @Resource
    private MqDeleteRetry mqDeleteRetry;

    public void delayDoubleDelete(String key, Runnable updateFunction){
        cacheDeleteRetry.delete(key);
        // 先更新
        updateFunction.run();
        // 再删除（重试）
        try {
            // 延迟500ms
            Thread.sleep(500);
            cacheDeleteRetry.delete(key);
        }catch (Exception e){
            log.error("delete cache failed！key={}",key, e);
            // 发送到mq删除（重试）
            try {
                mqDeleteRetry.sendDeleteMessage(key);
            }catch (Exception e1){
                // 发送到mq都失败了，不怕，key还有超时时间限制
                log.error("send delete message failed！key={}", key,e1);
            }
        }
    }

    @Component
    public static class CacheDeleteRetry {
        @Resource
        private RedisTemplate<String,String> redisTemplate;

        @Retryable(retryFor = RetryException.class, maxAttempts = 3, backoff = @Backoff(delay = 1000, multiplier = 1))
        public void delete(String key){
            Boolean deleted = redisTemplate.delete(key);
            if (Boolean.FALSE.equals(deleted)){
                throw new RetryException("deleted cache failed！");
            }
        }
    }

    @Component
    @RocketMQMessageListener(
            topic = MqDeleteRetry.TOPIC,
            consumerGroup = MqDeleteRetry.TOPIC
    )
    @Slf4j
    public static class MqDeleteRetry implements RocketMQListener<MessageExt> {
        private static final String TOPIC = "delete-cache-topic";

        @Resource
        private RocketMQTemplate rocketMQTemplate;

        @Resource
        private CacheDeleteRetry cacheDeleteRetry;

        @Retryable(retryFor = RetryException.class, maxAttempts = 3, backoff = @Backoff(delay = 1000, multiplier = 1))
        public void sendDeleteMessage(String key){
            // 发送到mq删除
            SendResult sendResult = rocketMQTemplate.syncSend("delete-cache-topic", key);
            if(sendResult.getSendStatus() != SendStatus.SEND_OK){
                throw new RetryException("send delete message failed！");
            }
        }
        @Override
        public void onMessage(MessageExt message) {
            byte[] body = message.getBody();
            cacheDeleteRetry.delete(new String(body, StandardCharsets.UTF_8));
        }

    }
}
